package com.nulldev.util.concurrency.threadIt.v4.utils.impls;

import java.util.Collection;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;

import com.nulldev.util.VariableAPI.ArrayUtils;
import com.nulldev.util.concurrency.threadIt.v4.asyncTasks.AsyncTaskCallback;
import com.nulldev.util.concurrency.threadIt.v4.emapi.ExecutorManager;
import com.nulldev.util.concurrency.threadIt.v4.emapi.uft.Executable;
import com.nulldev.util.data.Arrays.arrays.FastLists;
import com.nulldev.util.internal.backport.concurrency9.Lists;
import com.nulldev.util.logging.Logger;
import com.nulldev.util.logging.LoggerFactory;

public class AwaitTaskUtil {
	private static final Logger log = LoggerFactory.getLogger();
	/**
	 * Experiment: <br>
	 * Use Semaphores for synchronisation instead of CountdownLatches. <br>
	 * This might work. <br>
	 * <br>
	 * Current conclusion: <br>
	 * - CountdownLatches seem to be faster than semaphores. <br>
	 * - TODO: Investigate stability.
	 */
	private static final boolean ENBALE_SEMAPHORE_VAR = true;
	/**
	 * Use asynchronous callbacks instead of per-thread callbacks to improve latency
	 * and performance.
	 */
	private static final boolean USE_ASYNC_CALLBACK = true;

	private static final Collection<?> NULL_SINGLETONS = Collections.singleton(null);

	@SuppressWarnings(
		{ "unchecked", "rawtypes" })
	public static void await(final Executable<?>... t) {
		if (t == null || t.length == 0) {
			return;
		} else if (t.length == 1) {
			try {
				t[0].waitTillCompletion();
			} catch (Throwable e) {
				log.error(e);
			}
		} else if (!ENBALE_SEMAPHORE_VAR) {
			final CountDownLatch cdl = new CountDownLatch(t.length);
			log.debug("await(T: " + t.length + ")");
			final AsyncTaskCallback acb = USE_ASYNC_CALLBACK ? new AsyncTaskCallback<Object>() {

				@Override
				public void onCompletion(Object result) {
					cdl.countDown();
				}

				@Override
				public void onError(Throwable e) {
					cdl.countDown();
				}

				@Override
				public void onSubmissionRejection() {
					cdl.countDown();
				}

			} : null;
			for (final Executable<?> e : t) {
				if (USE_ASYNC_CALLBACK) {
					if (!e.addCallbackTask(acb)) {
						log.warn("Failed to add thread tracking! [task=" + e + "]");
						/* Make sure we account for not being able to track. */
						cdl.countDown();
					}
				} else {
					ExecutorManager.get().queueTask(() -> {
						log.debug("await(T: " + t.length + ", left: " + cdl.getCount() + ") E: WAITING for " + e);
						try {
							e.waitTillCompletion();
						} catch (Throwable ex) {
							log.error(ex);
						}
						log.debug("await(T: " + t.length + ", left: " + cdl.getCount() + ") E: " + e + " is done");
						cdl.countDown();
					});
				}
			}
			try {
				cdl.await();
			} catch (InterruptedException e1) {
				throw new RuntimeException(e1);
			}
		} else {
			final Semaphore sem = new Semaphore(0, false);
			log.debug("await(T: " + t.length + ")");
			final AsyncTaskCallback acb = USE_ASYNC_CALLBACK ? new AsyncTaskCallback<Object>() {

				@Override
				public void onCompletion(Object result) {
					sem.release(1);
				}

				@Override
				public void onError(Throwable e) {
					sem.release(1);
				}

				@Override
				public void onSubmissionRejection() {
					sem.release(1);
				}

			} : null;
			for (final Executable<?> e : t) {
				if (USE_ASYNC_CALLBACK) {
					if (!e.addCallbackTask(acb)) {
						log.warn("Failed to add thread tracking! [task=" + e + "]");
						/* Make sure we account for not being able to track. */
						sem.release(1);
					}
				} else {
					ExecutorManager.get().queueTask(() -> {
						log.debug("await(T: " + t.length + ", left: " + (t.length - sem.availablePermits()) + ") E: WAITING for " + e);
						try {
							e.waitTillCompletion();
						} catch (Throwable ex) {
							log.error(ex);
						}
						sem.release(1);
					});
				}
			}
			try {
				sem.acquire(t.length);
			} catch (InterruptedException e1) {
				throw new RuntimeException(e1);
			}
		}
	}

	@SuppressWarnings(
		{ "unchecked", "rawtypes" })
	private static List<Executable<?>> __await_for_cdl(final long amount, final TimeUnit unit, final Executable<?>... t) {
		final CountDownLatch cdl = new CountDownLatch(t.length);
		log.debug("await(T: " + t.length + ")");
		if (USE_ASYNC_CALLBACK) {
			final Executable<?>[] timedOut = new Executable<?>[t.length];
			System.arraycopy(t, 0, timedOut, 0, t.length);
			for (int i = 0; i < t.length; i++) {
				final int inx = i;
				final Executable<?> e = t[inx];
				if (!e.addCallbackTask((AsyncTaskCallback) new AsyncTaskCallback<Object>() {

					@Override
					public void onCompletion(Object result) {
						timedOut[inx] = null;
						cdl.countDown();
					}

					@Override
					public void onError(Throwable e) {
						timedOut[inx] = null;
						cdl.countDown();
					}

					@Override
					public void onSubmissionRejection() {
						timedOut[inx] = null;
						cdl.countDown();
					}

				})) {
					log.warn("Failed to add thread tracking! [task=" + e + "]");
					/* Make sure we account for not being able to track. */
					cdl.countDown();
				}
			}
			try {
				cdl.await(amount, unit);
			} catch (InterruptedException e1) {
				throw new RuntimeException(e1);
			}
			final List<Executable<?>> out = ArrayUtils.toArrayList(timedOut);
			out.removeAll(NULL_SINGLETONS);
			return out;
		} else {
			final List<Executable<?>> timedOut = FastLists.list(t.length);
			for (final Executable<?> e : t) {
				ExecutorManager.get().queueTask(() -> {
					log.debug("await(T: " + t.length + ", left: " + cdl.getCount() + ") E: WAITING for " + e);
					try {
						e.waitFor(amount, unit);
					} catch (Throwable ex) {
						timedOut.add(e);
					} finally {
						cdl.countDown();
					}
					log.debug("await(T: " + t.length + ", left: " + cdl.getCount() + ") E: " + e + " is done");
				});
			}
			try {
				cdl.await(amount, unit);
			} catch (InterruptedException e1) {
				throw new RuntimeException(e1);
			}
			return timedOut;
		}
	}

	@SuppressWarnings(
		{ "rawtypes", "unchecked" })
	private static List<Executable<?>> __await_for_semaphores(final long amount, final TimeUnit unit, final Executable<?>... t) {
		final Semaphore sem = new Semaphore(0, false);
		log.debug("await(T: " + t.length + ")");
		if (USE_ASYNC_CALLBACK) {
			final Executable<?>[] timedOut = new Executable<?>[t.length];
			System.arraycopy(t, 0, timedOut, 0, t.length);
			for (int i = 0; i < t.length; i++) {
				final int inx = i;
				final Executable<?> e = t[inx];
				if (!e.addCallbackTask((AsyncTaskCallback) new AsyncTaskCallback<Object>() {

					@Override
					public void onCompletion(Object result) {
						timedOut[inx] = null;
						sem.release(1);
					}

					@Override
					public void onError(Throwable e) {
						timedOut[inx] = null;
						sem.release(1);
					}

					@Override
					public void onSubmissionRejection() {
						timedOut[inx] = null;
						sem.release(1);
					}

				})) {
					log.warn("Failed to add thread tracking! [task=" + e + "]");
					/* Make sure we account for not being able to track. */
					sem.release(1);
				}
			}
			try {
				sem.tryAcquire(t.length, amount, unit);
			} catch (InterruptedException e1) {
				throw new RuntimeException(e1);
			}
			final List<Executable<?>> out = ArrayUtils.toArrayList(timedOut);
			out.removeAll(NULL_SINGLETONS);
			return out;
		} else {
			final List<Executable<?>> timedOut = FastLists.list(t);
			for (final Executable<?> e : t) {
				ExecutorManager.get().queueTask(() -> {
					log.debug("await(T: " + t.length + ", left: " + (t.length - sem.availablePermits()) + ") E: WAITING for " + e);
					try {
						e.waitFor(amount, unit);
					} catch (Throwable ex) {
						timedOut.add(e);
					} finally {
						sem.release(1);
					}
				});
			}
			try {
				sem.tryAcquire(t.length, amount, unit);
			} catch (InterruptedException e1) {
				throw new RuntimeException(e1);
			}
			return timedOut;
		}
	}

	public static List<Executable<?>> awaitFor(final long amount, final TimeUnit unit, final Executable<?>... t) {
		if (t == null || t.length == 0 || unit == null || amount <= 0) {
			return Lists.of();
		} else if (t.length == 1) {
			try {
				t[0].waitFor(amount, unit);
				return Lists.of();
			} catch (Throwable e) {
				return Lists.of(t[0]);
			}
		} else if (!ENBALE_SEMAPHORE_VAR) {
			return __await_for_cdl(amount, unit, t);
		} else {
			return __await_for_semaphores(amount, unit, t);
		}
	}

	@SuppressWarnings(
		{ "rawtypes", "unchecked" })
	public static void await(final List<Executable<?>> _t) {
		if (_t == null || _t.isEmpty()) {
			return;
		} else if (_t.size() == 1) {
			_t.get(0).waitTillCompletion();
		} else if (!ENBALE_SEMAPHORE_VAR) {
			final CountDownLatch cdl = new CountDownLatch(_t.size());
			log.debug("await(LT: " + _t.size() + ")");
			final AsyncTaskCallback acb = USE_ASYNC_CALLBACK ? new AsyncTaskCallback<Object>() {

				@Override
				public void onCompletion(Object result) {
					cdl.countDown();
				}

				@Override
				public void onError(Throwable e) {
					cdl.countDown();
				}

				@Override
				public void onSubmissionRejection() {
					cdl.countDown();
				}

			} : null;
			for (final Executable<?> e : _t) {
				if (USE_ASYNC_CALLBACK) {
					if (!e.addCallbackTask(acb)) {
						log.warn("Failed to add thread tracking! [task=" + e + "]");
						/* Make sure we account for not being able to track. */
						cdl.countDown();
					}
				} else {
					ExecutorManager.get().queueTask(() -> {
						log.debug("await(LT: " + _t.size() + ", left: " + cdl.getCount() + ") E: WAITING for " + e);
						try {
							e.waitTillCompletion();
						} catch (Throwable _ignored) {
						} finally {
							cdl.countDown();
						}
						log.debug("await(T: " + _t.size() + ", left: " + cdl.getCount() + ") E: " + e + " is done");
					});
				}
			}
			try {
				cdl.await();
			} catch (InterruptedException e1) {
				throw new RuntimeException(e1);
			}
		} else {
			final Semaphore sem = new Semaphore(0, false);
			log.debug("await(LT: " + _t.size() + ")");
			final AsyncTaskCallback acb = USE_ASYNC_CALLBACK ? new AsyncTaskCallback<Object>() {

				@Override
				public void onCompletion(Object result) {
					sem.release(1);
				}

				@Override
				public void onError(Throwable e) {
					sem.release(1);
				}

				@Override
				public void onSubmissionRejection() {
					sem.release(1);
				}

			} : null;
			for (final Executable<?> e : _t) {
				if (USE_ASYNC_CALLBACK) {
					if (!e.addCallbackTask(acb)) {
						log.warn("Failed to add thread tracking! [task=" + e + "]");
						/* Make sure we account for not being able to track. */
						sem.release(1);
					}
				} else {
					ExecutorManager.get().queueTask(() -> {
						log.debug("await(LT: " + _t.size() + ", left: " + (_t.size() - sem.availablePermits()) + ") E: WAITING for " + e);
						try {
							e.waitTillCompletion();
						} catch (Throwable _ignored) {
						} finally {
							sem.release(1);
						}
					});
				}
			}
			try {
				sem.acquire(_t.size());
			} catch (InterruptedException e1) {
				throw new RuntimeException(e1);
			}
		}
	}
}
